package com.atguigu.flink.chapter06.timer;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class TimerDemo1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env.socketTextStream("hadoop102",8888)
                .map(new WaterSensorMapFunction())
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    //当定时器触发的时候，会调用这个方法
                    @Override
                    public void onTimer(long timestamp,   //定时器时间
                                        OnTimerContext ctx,
                                        Collector<String> out) throws Exception {
                        out.collect("水位超过20，发出预警....." + timestamp);
                    }

                    long ts;
                    //处理元素
                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        //当水位的值大于20，5s后发出预警
                        if (value.getVc() > 20){
                            ts = ctx.timerService().currentProcessingTime();
                            //注册一个基于处理时间的  并且 5s后触发的定时器
                            ctx.timerService().registerProcessingTimeTimer(ts + 5000);
                        } else if (value.getVc() < 20) {
                            //取消定时器
                            ctx.timerService().deleteProcessingTimeTimer(ts + 5000);
                        }
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}



//public class TimerDemo1 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .keyBy(WaterSensor::getId)
//                /*
//                泛型1：输入数据流key的类型
//                泛型2：输入数据流中元素的类型
//                泛型3：输出的类型
//                 */
//                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
//                    //当定时器触发的时候，调用这个方法
//                    @Override
//                    public void onTimer(long timestamp,     //定时时间
//                                        OnTimerContext ctx,
//                                        Collector<String> out) throws Exception {
//                        out.collect("水位超过20，触发预警...." + timestamp);
//                    }
//
//                    long ts;
//                    //处理元素
//                    @Override
//                    public void processElement(WaterSensor value,
//                                               Context ctx,
//                                               Collector<String> out) throws Exception {
//                        //当水位值大于20,5s后触发定时器
//                        if (value.getVc() > 20){
//                            ts = ctx.timerService().currentProcessingTime();
//                            //注册一个基于处理时间 并且 5s后触发的定时器
//                            ctx.timerService().registerProcessingTimeTimer(ts + 5000);
//                        } else if (value.getVc() < 20) {
//                            //取消定时器
//                            ctx.timerService().deleteProcessingTimeTimer(ts + 5000);
//                        }
//                    }
//                })
//                .print();
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}


//public class TimerDemo1 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .keyBy(WaterSensor::getId)
//                /*
//                泛型1：输入的key的类型
//                泛型2：流中元素的类型
//                泛型3：输出的类型
//                 */
//                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
//                    //当定时器触发的时候，会调用这个方法
//                    @Override
//                    public void onTimer(long timestamp, //定时时间
//                                        OnTimerContext ctx,
//                                        Collector<String> out) throws Exception {
//                        out.collect("水位超过20，触发预警....." + timestamp);
//                    }
//
//                    long ts;
//                    @Override
//                    public void processElement(WaterSensor value,
//                                               Context ctx,
//                                               Collector<String> out) throws Exception {
//                        //当水位超过20，5s后触发定时器
//                        if (value.getVc() > 20){
//                            ts = ctx.timerService().currentProcessingTime();
//                            //注册一个基于处理时间的 并且 5s后触发的定时器
//                            ctx.timerService().registerProcessingTimeTimer(ts + 5000);
//                        } else if (value.getVc() < 20) {
//                            ctx.timerService().deleteProcessingTimeTimer(ts + 5000);
//                        }
//                    }
//                })
//                .print();
//
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}